package com.xdja.kafka.hdfs.sink.handle;

import com.xdja.kafka.hdfs.sink.bean.UserActiveBean;
import com.xdja.kafka.hdfs.sink.writer.definition.ParquetWriterDefinition;
import com.xdja.kafka.hdfs.sink.writer.manage.ParquetWriterManage;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.parquet.example.data.Group;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.*;

/**
 * 对kafka读取到的SinkRecords按日期分类以parquet格式保存到hadoop
 * 所在目录下的日期分类格式：YYYY-MM/YYYY-MM-DD/YYYY-MM-DD-*
 * Parquet格式文件不支持追加，对晚报的数据只能在YYYY-MM-DD文件夹下建立多个子文件，后期再通过job合并
 */
public class ParquetSinkHandle extends AbstractSinkHandle<ParquetWriterManage>{
    private static final Logger log = LoggerFactory.getLogger(ParquetSinkHandle.class);
    /**
     * 构造函数
     * @param sinkRecords
     */
    public ParquetSinkHandle(Collection<SinkRecord> sinkRecords) {
        super.sinkRecords = sinkRecords;
    }

    @Override
    protected void writeToHdfs(String date, List<UserActiveBean> userActiveBeanList) throws IOException {
        if(CollectionUtils.isEmpty(userActiveBeanList) || StringUtils.isEmpty(date)) {
            log.info("date或者userActiveBeanList为空，直接返回");
            return;
        }
        for(UserActiveBean userActiveBean : userActiveBeanList) {
            if(userActiveBean == null) {
                log.error("userActiveBean记录为null，忽略");
                continue;
            }
            //写入一条记录和换行符，每条记录占一行
            Group group = ParquetWriterDefinition.groupFactory.newGroup();
            if(!StringUtils.isEmpty(userActiveBean.getId())) {
                group.append("id", userActiveBean.getId());
            }
            if(!StringUtils.isEmpty(userActiveBean.getPoliceId())) {
                group.append("policeId", userActiveBean.getPoliceId());
            }
            if(!StringUtils.isEmpty(userActiveBean.getRequestId())) {
                group.append("requestId", userActiveBean.getRequestId());
            }
            if(!StringUtils.isEmpty(userActiveBean.getCardNo())) {
                group.append("cardNo", userActiveBean.getCardNo());
            }
            if(!StringUtils.isEmpty(userActiveBean.getSessionId())) {
                group.append("sessionId", userActiveBean.getSessionId());
            }
            if(!StringUtils.isEmpty(userActiveBean.getTerminalIp())) {
                group.append("terminalIp", userActiveBean.getTerminalIp());
            }
            if(!StringUtils.isEmpty(userActiveBean.getSn())) {
                group.append("sn", userActiveBean.getSn());
            }
            if(!StringUtils.isEmpty(userActiveBean.getSource())) {
                group.append("source", userActiveBean.getSource());
            }
            if(!StringUtils.isEmpty(userActiveBean.getLogType())) {
                group.append("logType", userActiveBean.getLogType());
            }
            if(!StringUtils.isEmpty(userActiveBean.getBizType())) {
                group.append("bizType", userActiveBean.getBizType());
            }
            if(!StringUtils.isEmpty(userActiveBean.getModule())) {
                group.append("module", userActiveBean.getModule());
            }
            if(!StringUtils.isEmpty(userActiveBean.getParams())) {
                group.append("params", userActiveBean.getParams());
            }
            if(!StringUtils.isEmpty(userActiveBean.getFormatParam())) {
                group.append("formatParam", userActiveBean.getFormatParam());
            }
            if(!StringUtils.isEmpty(userActiveBean.getUrl())) {
                group.append("url", userActiveBean.getUrl());
            }
            if(!StringUtils.isEmpty(userActiveBean.getUri())) {
                group.append("uri", userActiveBean.getUri());
            }
            if(!StringUtils.isEmpty(userActiveBean.getContent())) {
                group.append("content", userActiveBean.getContent());
            }
            if(!StringUtils.isEmpty(userActiveBean.getResponse())) {
                group.append("response", userActiveBean.getResponse());
            }
            if(!StringUtils.isEmpty(userActiveBean.getResponseType())) {
                group.append("responseType", userActiveBean.getResponseType());
            }
            if(userActiveBean.getResponseTime() != null) {
                group.append("responseTime", userActiveBean.getResponseTime());
            }
            if(!StringUtils.isEmpty(userActiveBean.getResult())) {
                group.append("result", userActiveBean.getResult());
            }
            if(!StringUtils.isEmpty(userActiveBean.getSourceIp())) {
                group.append("sourceIp", userActiveBean.getSourceIp());
            }
            if(userActiveBean.getSourcePort() != null) {
                group.append("sourcePort", userActiveBean.getSourcePort());
            }
            if(!StringUtils.isEmpty(userActiveBean.getDestIp())) {
                group.append("destIp", userActiveBean.getDestIp());
            }
            if(userActiveBean.getDestPort() != null) {
                group.append("destPort", userActiveBean.getDestPort());
            }
            if(userActiveBean.getTime() != null) {
                group.append("time", userActiveBean.getTime());
            }
            if(!StringUtils.isEmpty(userActiveBean.getErrorCode())) {
                group.append("errorCode", userActiveBean.getErrorCode());
            }
            if(!StringUtils.isEmpty(userActiveBean.getErrorLog())) {
                group.append("errorLog", userActiveBean.getErrorLog());
            }
            if(!StringUtils.isEmpty(userActiveBean.getSystemId())) {
                group.append("systemId", userActiveBean.getSystemId());
            }
            if(!StringUtils.isEmpty(userActiveBean.getNetworkCode())) {
                group.append("networkCode", userActiveBean.getNetworkCode());
            }
            if(!StringUtils.isEmpty(userActiveBean.getSystemName())) {
                group.append("systemName", userActiveBean.getSystemName());
            }
            if(!StringUtils.isEmpty(userActiveBean.getNetworkName())) {
                group.append("networkName", userActiveBean.getNetworkName());
            }
            if(!StringUtils.isEmpty(userActiveBean.getKeyWords())) {
                group.append("keyWords", userActiveBean.getKeyWords());
            }
            ParquetWriterManage.getInstance().getWriter(date).write(group);
        }
    }
}
